﻿using Microsoft.Extensions.Hosting;
using System.Text;
using Microsoft.Extensions.Logging;
using RabbitMQ.Client.Events;
using RabbitMQ.Client;
using Core.RabbitMQ.Attribute;
using System.Reflection;
using Newtonsoft.Json;

namespace Core.RabbitMQ.Consumer
{
    public class RabbitMqConsumerService<TMessage, TConsumer> : BackgroundService
        where TMessage : class, new()
        where TConsumer : IConsumer<TMessage>
    {
        /// <summary>
        /// 消费队列配置参数
        /// </summary>
        private ConsumerOptions Options { get; set; }

        private readonly IConnection _connection;

        private readonly IModel _consumerModel;

        private AsyncEventingBasicConsumer _asyncConsumer;

        private readonly ILogger<TConsumer> _logger;

        private readonly TConsumer _consumer;

        /// <summary>
        /// 
        /// </summary>
        /// <param name="connectionProxy"></param>
        /// <param name="serviceProvider"></param>
        /// <param name="logger"></param>
        public RabbitMqConsumerService(IRabbitMqConnectionProxy connectionProxy,
            ILogger<TConsumer> logger,
            TConsumer consumer)
        {
            _logger = logger;
            _connection = connectionProxy.CreateConnection();
            _consumerModel = _connection.CreateModel();
            LoadOptions();
            _consumer = consumer;
        }

        public override Task StartAsync(CancellationToken cancellationToken)
        {
            return base.StartAsync(cancellationToken);
        }

        private CancellationToken _stoppingToken;
        protected override Task ExecuteAsync(CancellationToken stoppingToken)
        {
            _stoppingToken = stoppingToken;
            ConnectionReady();
            _asyncConsumer = new AsyncEventingBasicConsumer(_consumerModel);
            _asyncConsumer.Received -= ConsumerReceivedAsync;
            _asyncConsumer.Received += ConsumerReceivedAsync;

            var consumeTag = _consumerModel.BasicConsume(
                queue: Options.QueueName,
                autoAck: false,
                consumerTag: Environment.MachineName,
                consumer: _asyncConsumer
                );
            _asyncConsumer.HandleBasicCancelOk(consumeTag);
            return Task.CompletedTask;
        }

        /// <summary>
        /// 接收处理消息，完成后确认
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="eventArgs"></param>
        /// <returns></returns>
        private async Task ConsumerReceivedAsync(object? sender, BasicDeliverEventArgs eventArgs)
        {
            try
            {
                var isFired = await FireReceiveEventAsync(eventArgs);

                if (isFired)
                {
                    _consumerModel.BasicAck(eventArgs.DeliveryTag, false);
                }
                else
                {
                    _consumerModel.BasicNack(eventArgs.DeliveryTag, false, false);
                }
            }
            catch (Exception e)
            {
                _logger.LogError(e, $"[{Options.QueueName}]: RabbitMQ Consumer Ack Exception!");
            }
        }

        /// <summary>
        /// 消息处理
        /// </summary>
        /// <param name="messageId"></param>
        /// <param name="body"></param>
        /// <returns></returns>
        private async Task<bool> FireReceiveEventAsync(BasicDeliverEventArgs eventArgs)
        {
            try
            {
                var messageStr = Encoding.UTF8.GetString(eventArgs.Body.ToArray());
                var message = JsonConvert.DeserializeObject<TMessage>(messageStr);
                await _consumer.ReceiveMessageAsync(eventArgs.BasicProperties.MessageId, message,_stoppingToken);
                return true;
            }
            catch (Exception e)
            {
                _logger.LogError(e, $"[{Options.QueueName}]: RabbitMQ Consumer FireReceiveEvent Exception");
                return false;
            }
        }

        /// <summary>
        /// 停止执行
        /// </summary>
        /// <param name="cancellationToken"></param>
        /// <returns></returns>
        public override Task StopAsync(CancellationToken cancellationToken)
        {
            _asyncConsumer.Received -= ConsumerReceivedAsync;
            _consumerModel.Close();
            _consumerModel.Dispose();
            _connection.Close();
            _connection.Dispose();
            _logger.LogInformation($"[{Options.QueueName}]: RabbitMQ Consumer is Stoped!");
            return base.StopAsync(cancellationToken);
        }
        /// <summary>
        /// 加载配置
        /// </summary>
        /// <exception cref="Exception"></exception>
        private void LoadOptions()
        {
            var type = typeof(TMessage);
            var mqConsumeAttribut = type.GetCustomAttribute<RabbitMqMessageAttribute>();
            if (mqConsumeAttribut == null)
            {
                throw new Exception($"消息类型[{typeof(TMessage).Name}]未配置RabbitMqMessage特性！");
            }
            Options = new ConsumerOptions
            {
                QueueName = mqConsumeAttribut.QueueName,
                Exchange = mqConsumeAttribut.Exchange,
                ExchangeType = mqConsumeAttribut.ExchangeType,
                RoutingKey = mqConsumeAttribut.RoutingKey,
                PrefetchCount = mqConsumeAttribut.PrefetchCount,
                QueueTtl = mqConsumeAttribut.QueueTtl
            };
        }

        /// <summary>
        /// 准备连接
        /// </summary>
        /// <exception cref="Exception"></exception>
        private void ConnectionReady()
        {
            if (string.IsNullOrEmpty(Options.QueueName))
            {
                throw new Exception("请在Options中设置发送消息的目标队列名称！");
            }

            //定义死信交换机
            _consumerModel.ExchangeDeclare(
                exchange: $"{Options.QueueName}_Dead_Exchange",
                durable: true,
                type: Options.ExchangeType.ToString().ToLower(),
                autoDelete: false
            );
            //定义死信队列
            _consumerModel.QueueDeclare(
                queue: $"{Options.QueueName}_Dead_Queue",
                durable: true,
                exclusive: false,
                autoDelete: false
            );
            //绑定死信交换机和队列
            _consumerModel.QueueBind(
                queue: $"{Options.QueueName}_Dead_Queue",
                exchange: $"{Options.QueueName}_Dead_Exchange",
                routingKey: $"{Options.QueueName}_Dead_RoutingKey"
            );

            var consumerQueueArgument = new Dictionary<string, object>
            {
                { Headers.XDeadLetterExchange, $"{Options.QueueName}_Dead_Exchange" },
                { Headers.XMessageTTL, Options.QueueTtl },
                { Headers.XDeadLetterRoutingKey, $"{Options.QueueName}_Dead_RoutingKey" }
            };

            var exchangeName = string.IsNullOrEmpty(Options.Exchange) ? $"{Options.QueueName}_Exchange" : Options.Exchange;
            var routingKey = string.IsNullOrEmpty(Options.RoutingKey) ? $"{Options.QueueName}_RoutingKey" : Options.RoutingKey;
            //定义交换机
            _consumerModel.ExchangeDeclare(
                exchange: exchangeName,
                durable: true,
                type: Options.ExchangeType.ToString().ToLower(),
                autoDelete: false
            );
            //定义队列
            _consumerModel.QueueDeclare(
                queue: Options.QueueName,
                durable: true,
                exclusive: false,
                autoDelete: false,
                consumerQueueArgument
            );
            //绑定交换机和队列
            _consumerModel.QueueBind(
                queue: Options.QueueName,
                exchange: exchangeName,
                routingKey: routingKey
            );

        }
    }
}
